package com.johnguo.carbondata_demo;

import org.apache.commons.lang.math.NumberUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.CarbonContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SaveMode;

public class InsertData {
	public static void main(String... args) {
		SparkConf conf = new SparkConf().setAppName("insertData");
		SparkContext sc = new SparkContext(conf);
		JavaSparkContext jsc = new JavaSparkContext(sc);
		CarbonContext cc = new CarbonContext(sc, "hdfs://master:9000/usr/cardata");
		JavaRDD rdd = jsc.textFile("/test/sample.csv").map(new Function<String, TestTable>() {

			@Override
			public TestTable call(String record) throws Exception {
				String[] val = record.split(",");
				TestTable tt = new TestTable();
				tt.setName(val[0]);
				tt.setCity(val[1]);
				if (NumberUtils.isNumber(val[2])) {
					tt.setAge(Integer.valueOf(val[2]));
				}
				return tt;
			}
		});
		DataFrame df = cc.createDataFrame(rdd, TestTable.class);
		df.write().format("org.apache.spark.sql.CarbonSource").option("tableName", "test_table")
				.mode(SaveMode.Append).save();

	}
}
